package mysql

import (
	"database/sql"
	"errors"
	"fmt"
	_ "github.com/go-sql-driver/mysql"
	"github.com/sirupsen/logrus"
	"miniDW/src/datapack"
	"miniDW/src/log"
	"strconv"
	"strings"
)

const (
	MAX_ROWSBUFFER   = 400
	METHOD_OVERWRITE = "overwrite"
)

func getUrl(m map[string]interface{}) string {
	if v, ok := m["url"].(string); ok {
		return v
	}
	return ""
}

func getDatabase(m map[string]interface{}) string {
	if v, ok := m["database"].(string); ok {
		return v
	}
	return ""
}

func getPort(m map[string]interface{}) int {
	if v, ok := m["port"].(int); ok {
		return v
	}
	return 3306
}

func getUser(m map[string]interface{}) string {
	if v, ok := m["user"].(string); ok {
		return v
	}
	return ""
}

func getPassword(m map[string]interface{}) string {
	if v, ok := m["password"].(string); ok {
		return v
	}
	return ""
}

type MysqlConfig struct {
	Host     string
	Database string
	Port     int
	User     string
	Password string
}

func (s *MysqlConfig) String() string {
	return fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4&parseTime=True&loc=Local",
		s.User, s.Password, s.Host, s.Port, s.Database)
}

func Connect(config *MysqlConfig) (*sql.DB, error) {
	log.Debug(config.String())
	db, err := sql.Open("mysql", config.String())
	if err != nil {
		return nil, log.Error(err)
	}
	err = db.Ping()
	if err != nil {
		return nil, log.Error(err)
	}
	return db, nil
}

func parseConfig(data interface{}) (*MysqlConfig, error) {
	c := MysqlConfig{}
	if m, ok := data.(map[string]interface{}); ok {
		logrus.Debug(m)
		c.Host = getUrl(m)
		if c.Host == "" {
			return nil, log.Error(errors.New("host is empty"))
		}
		c.Database = getDatabase(m)
		if c.Database == "" {
			return nil, log.Error(errors.New("database is empty"))
		}
		c.Port = getPort(m)
		if c.Port == 0 {
			c.Port = 3306
		}
		c.User = getUser(m)
		if c.User == "" {
			return nil, log.Error(errors.New("user is empty"))
		}
		c.Password = getPassword(m)
		if c.Password == "" {
			return nil, log.Error(errors.New("password is empty"))
		}
	} else {
		return nil, log.Error(errors.New("data is not map[string]interface{}"))
	}
	return &c, nil
}

func splitTypeWords(data string) []string {
	i := 0
	word := ""
	words := make([]string, 0)
	for i < len(data) {
		if (data[i] >= 'a' && data[i] <= 'z') ||
			(data[i] >= 'A' && data[i] <= 'Z') ||
			(data[i] >= '0' && data[i] <= '9') {
			word += string(data[i])
		}
		if data[i] == '(' || data[i] == ')' || data[i] == ',' {
			words = append(words, word)
			words = append(words, string(data[i]))
			word = ""
		}
		if data[i] == ' ' || data[i] == '\t' {
			words = append(words, word)
			word = ""
		}
		i++
	}
	if word != "" {
		words = append(words, word)
	}
	return words
}

func newFieldType(name string, dataType string) (*datapack.FieldType, error) {
	dataType = strings.ToLower(dataType)
	words := splitTypeWords(dataType)
	if len(words) == 0 {
		return nil, fmt.Errorf("invalid data type: %s", dataType)
	}

	field := &datapack.FieldType{
		Name: name,
	}

	switch words[0] {
	case "int":
		field.Type = datapack.FIELDTYPE_INTEGER
		field.Length = 1
		field.FloatLength = 0
		field.ByteLength = 8
		return field, nil
	case "bigint":
		field.Length = 1
		field.FloatLength = 0
		field.ByteLength = 8
		if len(words) == 1 {
			field.Type = datapack.FIELDTYPE_BIGINT
			return field, nil
		}
		if words[1] == "unsigned" {
			field.Type = datapack.FIELDTYPE_UNSIGNEDBIGINT
			return field, nil
		}
	case "unsigned":
		switch words[1] {
		case "bigint":
			field.Length = 1
			field.FloatLength = 0
			field.ByteLength = 8
			if len(words) == 1 {
				field.Type = datapack.FIELDTYPE_UNSIGNEDBIGINT
				return field, nil
			}
		default:
			for _, word := range words {
				fmt.Println(word)
			}
			return nil, log.Error(fmt.Errorf("data type %s not support", dataType))
		}
	case "timestamp":
		field.Type = datapack.FIELDTYPE_TIMESTAMP
		field.ByteLength = 20
		return field, nil
	case "datetime":
		field.Type = datapack.FIELDTYPE_DATETIME
		field.ByteLength = 20
	case "varchar":
		field.Type = datapack.FIELDTYPE_STRING
		field.ByteLength = 1
		if len(words) == 1 {
			return field, nil
		}
		if len(words) >= 4 && words[1] == "(" {
			length, err := strconv.Atoi(words[2])
			if err != nil {
				return nil, log.Error(err)
			}
			field.Length = uint16(length)
			field.ByteLength = length
			return field, nil
		}
	case "text":
		field.Type = datapack.FIELDTYPE_TEXT
		field.ByteLength = 256
		return field, nil
	case "decimal":
		field.Type = datapack.FIELDTYPE_FLOAT
		field.ByteLength = 8
		if len(words) == 1 {
			return field, nil
		}
		if len(words) >= 4 && words[1] == "(" {
			length, err := strconv.Atoi(words[2])
			if err != nil {
				return nil, log.Error(err)
			}
			field.Length = uint16(length)
			field.ByteLength = length
			if words[3] == "," {
				if len(words) > 5 {
					floatLength, err := strconv.Atoi(words[4])
					if err != nil {
						return nil, err
					}
					field.FloatLength = uint16(floatLength)
					field.ByteLength += floatLength + 1
				}
				return field, nil
			}
		}
		return field, nil
	default:
		for _, word := range words {
			fmt.Println(word)
		}
		return nil, log.Error(fmt.Errorf("data type %s not support", dataType))
	}

	return field, nil
}

func BuildDataType(field *datapack.FieldType) string {
	switch field.Type {
	case datapack.FIELDTYPE_INTEGER:
		return "int"
	case datapack.FIELDTYPE_NUMBER:
		return "bigint"
	case datapack.FIEKDTYPE_UNSIGNEDNUMBER:
		return "bigint unsigned"
	case datapack.FIELDTYPE_TIMESTAMP:
		return "timestamp"
	case datapack.FIELDTYPE_STRING:
		if field.Length > 0 {
			return "varchar(" + strconv.Itoa(int(field.Length)) + ")"
		}
		return "text"
	case datapack.FIELDTYPE_FLOAT:
		if field.Length > 0 {
			return "decimal(" + strconv.Itoa(int(field.Length)) + "," +
				strconv.Itoa(int(field.FloatLength)) + ")"
		}
		return "float"
	default:
		return "text"
	}
}

func CreateTableStructure(db *sql.DB, tableStruct *datapack.TableStruct, method string) error {
	log.Debug("start create table ", tableStruct.Name)
	var columnDefinitions []string
	var colDef string

	if method == METHOD_OVERWRITE {
		_, err := db.Exec("DROP TABLE IF EXISTS `" + tableStruct.Name() + "`")
		if err != nil {
			return err
		}
	}

	names := tableStruct.GetFeildsNames()
	for _, name := range names {
		//fmt.Println(name)
		field := tableStruct.GetFieldTypeByName(name)
		if field.Type == datapack.FIELDTYPE_TIMESTAMP {
			colDef = fmt.Sprintf("`%s` %s NULL DEFAULT NULL", field.Name, BuildDataType(field))
		} else {
			colDef = fmt.Sprintf("`%s` %s", field.Name, BuildDataType(field))
		}
		columnDefinitions = append(columnDefinitions, colDef)
	}

	createTableSQL := fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s` (\n%s\n);", tableStruct.Name(),
		// 使用逗号拼接各列定义字符串
		strings.Join(columnDefinitions, ",\n"))
	//fmt.Println(createTableSQL)
	_, err := db.Exec(createTableSQL)
	if err != nil {
		return log.Error(err, createTableSQL)
	}
	return nil
}

// 获取所有表名
func getTableNames(db *sql.DB) ([]string, error) {
	var tableNames []string
	rows, err := db.Query("SHOW TABLES")
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var tableName string
		if err := rows.Scan(&tableName); err != nil {
			return nil, err
		}
		tableNames = append(tableNames, tableName)
	}
	return tableNames, nil
}

// 创建表结构
func createTableStructure(sourceDB, targetDB *sql.DB, tableName string) error {
	var createTableSQL string
	row := sourceDB.QueryRow(fmt.Sprintf("SHOW CREATE TABLE %s", tableName))
	if err := row.Scan(&tableName, &createTableSQL); err != nil {
		return err
	}
	_, err := targetDB.Exec(createTableSQL)
	return err
}

// 拷贝表数据
func copyTableData(sourceDB, targetDB *sql.DB, tableName string) error {
	columns, err := getColumns(sourceDB, tableName)
	if err != nil {
		return err
	}
	columnNames := make([]string, len(columns))
	for i, col := range columns {
		columnNames[i] = col.ColumnName
	}
	query := fmt.Sprintf("SELECT %s FROM %s", strings.Join(columnNames, ","), tableName)
	rows, err := sourceDB.Query(query)
	if err != nil {
		return err
	}
	defer rows.Close()
	// 准备插入语句
	insertSQL := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", tableName, strings.Join(columnNames, ","),
		strings.TrimSuffix(strings.Repeat("?,", len(columnNames)), ","))
	stmt, err := targetDB.Prepare(insertSQL)
	if err != nil {
		return err
	}
	defer stmt.Close()
	for rows.Next() {
		values := make([]interface{}, len(columns))
		valuePtrs := make([]interface{}, len(columns))
		for i := range columns {
			valuePtrs[i] = &values[i]
		}
		if err := rows.Scan(valuePtrs...); err != nil {
			return err
		}
		_, err := stmt.Exec(values...)
		if err != nil {
			return err
		}
	}
	return nil
}

// 获取表的列信息
func getColumns(db *sql.DB, tableName string) ([]ColumnInfo, error) {
	var columns []ColumnInfo
	rows, err := db.Query(fmt.Sprintf("DESCRIBE %s", tableName))
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var col ColumnInfo
		if err := rows.Scan(&col.ColumnName, &col.DataType, &col.Nullable, &col.Key, &col.DefaultValue, &col.Extra); err != nil {
			return nil, log.Error(err)
		}
		columns = append(columns, col)
	}
	return columns, nil
}

// 列信息结构体
type ColumnInfo struct {
	ColumnName   string
	DataType     string
	Nullable     string
	Key          string
	DefaultValue interface{}
	Extra        string
}
